package ontime

import java.util

import org.apache.commons.io.FileUtils
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment

/**
  * @author dinghao 
  * @create 2021-07-28 14:49 
  * @message
  */
object Cacheflink {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.registerCachedFile(args(1), "dh")

//    val input = env.fromElements("a","b","c")
    val input = env.socketTextStream(args(0),9999)

    val list = new util.ArrayList[String]()

    input.map(new RichMapFunction[String, String] {
      override def open(parameters: Configuration): Unit = {
        val dcFile = getRuntimeContext.getDistributedCache.getFile("dh")
        val lines = FileUtils.readLines(dcFile)
        import scala.collection.JavaConverters._
        for(ele <- lines.asScala){
          println("============================================"+ele)
          list.add(ele)
        }

      }
      override def map(value: String) = {
        value+"================"+util.Arrays.asList(list).toString
      }
    }).print()
    env.execute("Cacheflink")
  }
}
